package com.hx.vending.util;

import com.hx.vending.mqtt.ClientMQTT;
import org.apache.log4j.Logger;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

import javax.net.ssl.SSLSocketFactory;
import java.util.UUID;

public class PubMsg {
	private static int qos = 0; //只有一次

    public static final String broker = "ssl://47.105.178.86:9090";
//    private static String broker =  "tcp://47.105.178.86:1883";
    private static String userName = "admin";
    private static String passWord = "public";
    private static String threadId;  

    private static ClientMQTT clientMQTT = null;
   
    private static Logger logger=Logger.getLogger(PubMsg.class); // 获取logger实例
    
    private static MqttClient connect(String clientId,String userName, String password) throws Exception {

        MemoryPersistence persistence = new MemoryPersistence();  
        MqttConnectOptions connOpts = new MqttConnectOptions();

        connOpts.setCleanSession(true);  
        connOpts.setUserName(userName);  
        connOpts.setPassword(password.toCharArray());  
        connOpts.setConnectionTimeout(10);  
        connOpts.setKeepAliveInterval(20);
        //String[] uris = {"tcp://10.100.124.206:1883","tcp://10.100.124.207:1883"};
        //connOpts.setServerURIs(uris);  //起到负载均衡和高可用的作用
        MqttClient mqttClient = new MqttClient(broker, clientId, persistence);  
        mqttClient.setCallback(new MqttCallback() {
			
			@Override
			public void messageArrived(String topic, MqttMessage message) throws Exception {
				String msg = new String(message.getPayload());
				logger.info(threadId + " " + msg);
	          
			}
			
			@Override
			public void deliveryComplete(IMqttDeliveryToken token) {
				// TODO Auto-generated method stub
				logger.info("deliveryComplete---------" + token.isComplete()); 
			}
			
			@Override
			public void connectionLost(Throwable arg0) {
				logger.info("断开。。。。");
				
			}
		});
        if (clientMQTT == null){
            clientMQTT = new ClientMQTT();
        }
//        System.out.println("-----------------------------------------" +ClientMQTT.path);
        //证书路径
        SSLSocketFactory sslSocketFactory = clientMQTT.getSSLSocktet(ClientMQTT.path + "ca.crt", ClientMQTT.path + "client.crt",ClientMQTT.path + "client.pem", "");
        //
        connOpts.setSocketFactory(sslSocketFactory);

        mqttClient.connect(connOpts);

//        boolean boo = true;
//        while(boo){
//            try {
//                mqttClient.connect(connOpts);
//                boo = false;
//            } catch (Exception e) {
//                boo = true;
//                System.out.println("------------------------发布主题连接失败，10秒后重连-------------------------");
//                Thread.sleep(10000);
//            }
//        }
        return mqttClient;
    }
      
    public static void publish(String str,String topic) throws Exception {
	    publish(str, UUID.randomUUID().toString().replace("-", ""), topic);
    }
    
    public static void publish(String str,String clientId,String topic) throws Exception {
        MqttClient mqttClient = connect(clientId,userName,passWord);  
  
        if (mqttClient != null) {  
            pub(mqttClient, str, topic);  
            logger.info("pub-->" + str);
        }  
  
        if (mqttClient != null) {  
            mqttClient.disconnect();  
        }  
    }

    private static void pub(MqttClient sampleClient, String msg,String topic)throws MqttPersistenceException, MqttException {
        MqttMessage message = new MqttMessage(msg.getBytes());
        message.setQos(qos);
        message.setRetained(false);
        sampleClient.publish(topic, message);
    }

}
